package com.qudian.pay.schedule.service;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.qudian.pay.core.utils.IPUtils;
import com.qudian.pay.schedule.client.TaskService;
import com.qudian.pay.schedule.enums.TaskRunningStatusEntum;
import com.qudian.pay.schedule.model.TaskModel;
import com.qudian.pay.schedule.util.Constant;

/**
 * @Description: 任务调度客户端服务
 * @ClassName: ScheduleClient
 * @author gaoxiang
 * @date 2016年7月1日 下午4:02:03
 */ 
public class ScheduleClient {
	
	private static Logger logger = LoggerFactory.getLogger(ScheduleClient.class);
	
	@Autowired
	private IScheduleService scheduleService;
	
	/**
	 * @Fields group : 分组标识 代表一种任务，在任务分配的时候用到
	 */ 
	private String group;
	
	/**
	 * @Fields tasks : 任务列表
	 */ 
	private List<TaskService> tasks;
	
	/**
	 * @Fields uuid : 唯一标识
	 */ 
	private UUID uuid;
	
	/**
	 * @Fields shutdownFlag : 服务关闭标识
	 */ 
	private volatile boolean shutdownFlag;
	
	/**
	 * @Fields executor : 线程池
	 */ 
	private Executor executor = Executors.newSingleThreadExecutor();
	
	private List<TaskModel> taskModelList = Lists.newArrayList();
	
	private Map<String, TaskService> taskMap = Maps.newHashMap();
	
	/**
	 * @Description: 初始化服务
	 * @return void 返回类型
	 * @author gaoxiang
	 * @date 2016年7月1日 下午4:06:48
	 */
	public void init() {
		shutdownFlag =false;
		this.uuid = UUID.randomUUID();
		if (tasks != null && !tasks.isEmpty()) {
			//开启监控线程
			new TaskExpireTimerThread().start();
			for (TaskService task : tasks) {
				TaskModel taskModel = new TaskModel();
				taskModel.setGroup(group);
				taskModel.setVersion(task.getVersion());
				taskModel.setName(task.getClass().getName());
				taskModel.setUuid(uuid.toString());
				taskModel.setIp(IPUtils.getIp4Single());
				taskModel.setStatus(TaskRunningStatusEntum.INIT.toString());
				taskModel.setTaskTag(task.getTaskTag());
				taskModelList.add(taskModel);
				//任务集合
				taskMap.put(task.getTaskTag(), task);
			}
			//客户端结点注册
			scheduleService.registerSchedule(taskModelList);
			executor.execute(new Runnable() {
				public void run() {
					//消息订阅
					scheduleService.subscribe(group, uuid.toString(), taskMap);
				}
			});
		}
	}
	
	/**
	 * @Description: 关闭的时候销毁线程
	 * @return void 返回类型
	 * @author gaoxiang
	 * @date 2016年5月19日 下午1:04:56
	 */
	public void destroy(){
		shutdownFlag =true;
	}
	
	public void updateExpireTime(List<TaskModel> taskModels) {
		if ((tasks != null) && (!tasks.isEmpty()))
			scheduleService.updateExpireTime(taskModels);
	}
	
	public void setTasks(List<TaskService> tasks) {
		this.tasks = tasks;
	}
	
	public void setGroup(String group) {
		this.group = group;
	}
	
	class TaskExpireTimerThread extends Thread {
		TaskExpireTimerThread(){
			setDaemon(true);
		}
		public void run() {
			if ((tasks != null) && (!tasks.isEmpty())){
				while (!shutdownFlag){
					try {
						Thread.sleep(Constant.CLIENT_SKIP_TIMES);
						scheduleService.checkChannelAlive(group, uuid.toString(), taskMap);
						logger.debug("update expire time ...." + new Date());
						scheduleService.updateExpireTime(taskModelList);
					} catch (Throwable t) {
						logger.error("TaskExpireTimerThread exception", t);
					}
				}
				//清除 redis缓存
				if(shutdownFlag){
					scheduleService.clearCache(uuid);
				}
			}
		}
	}
}
